//! 数据一致性协议实现

use crate::sync::{DataSyncManager, SyncMessage};
use crate::node::ClusterNodeInfo;
use crate::manager::ClusterManager;
use nacos_core::{Result, NacosError};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use chrono::{DateTime, Utc};
use tracing::{info, warn};
use uuid::Uuid;

/// 一致性协议类型
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub enum ConsistencyProtocol {
    /// 强一致性（Raft）
    Strong,
    /// 最终一致性（Gossip）
    Eventual,
    /// 混合一致性
    Hybrid,
}

/// 一致性级别
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
pub enum ConsistencyLevel {
    /// 强一致性
    Strong,
    /// 最终一致性
    Eventual,
    /// 读已提交
    ReadCommitted,
    /// 读未提交
    ReadUncommitted,
}

/// 投票状态
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum VoteStatus {
    /// 待投票
    Pending,
    /// 已接受
    Accepted,
    /// 已拒绝
    Rejected(String),
    /// 超时
    Timeout,
}

/// 投票请求
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoteRequest {
    /// 投票ID
    pub vote_id: String,
    /// 提案ID
    pub proposal_id: String,
    /// 源节点ID
    pub source_node_id: String,
    /// 数据版本
    pub version: u64,
    /// 时间戳
    pub timestamp: DateTime<Utc>,
    /// 数据哈希
    pub data_hash: u64,
}

/// 投票响应
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VoteResponse {
    /// 投票ID
    pub vote_id: String,
    /// 响应节点ID
    pub responder_id: String,
    /// 投票状态
    pub status: VoteStatus,
    /// 响应时间
    pub response_time: DateTime<Utc>,
}

/// 提案状态
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ProposalStatus {
    /// 待处理
    Pending,
    /// 已接受
    Accepted,
    /// 已拒绝
    Rejected(String),
    /// 已提交
    Committed,
    /// 已回滚
    RolledBack,
}

/// 数据提案
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DataProposal {
    /// 提案ID
    pub proposal_id: String,
    /// 提案节点ID
    pub proposer_id: String,
    /// 同步消息
    pub sync_message: SyncMessage,
    /// 提案状态
    pub status: ProposalStatus,
    /// 投票计数
    pub vote_count: u32,
    /// 需要的最小投票数
    pub required_votes: u32,
    /// 创建时间
    pub created_at: DateTime<Utc>,
    /// 过期时间
    pub expires_at: DateTime<Utc>,
    /// 投票记录
    pub votes: HashMap<String, VoteStatus>,
}

impl DataProposal {
    /// 创建新的数据提案
    pub fn new(
        proposer_id: String,
        sync_message: SyncMessage,
        required_votes: u32,
        timeout_seconds: i64,
    ) -> Self {
        let proposal_id = Uuid::new_v4().to_string();
        let now = Utc::now();
        
        Self {
            proposal_id,
            proposer_id,
            sync_message,
            status: ProposalStatus::Pending,
            vote_count: 0,
            required_votes,
            created_at: now,
            expires_at: now + chrono::Duration::seconds(timeout_seconds),
            votes: HashMap::new(),
        }
    }

    /// 检查是否过期
    pub fn is_expired(&self) -> bool {
        Utc::now() > self.expires_at
    }

    /// 添加投票
    pub fn add_vote(&mut self, node_id: String, status: VoteStatus) {
        self.votes.insert(node_id, status);
        self.update_vote_count();
    }

    /// 更新投票计数
    fn update_vote_count(&mut self) {
        self.vote_count = self.votes.values()
            .filter(|status| matches!(status, VoteStatus::Accepted))
            .count() as u32;
    }

    /// 检查是否获得足够投票
    pub fn has_majority(&self) -> bool {
        self.vote_count >= self.required_votes
    }

    /// 检查是否可以提交
    pub fn can_commit(&self) -> bool {
        self.has_majority() && !self.is_expired()
    }
}

/// 一致性管理器
#[derive(Debug)]
pub struct ConsistencyManager {
    /// 数据同步管理器
    data_sync_manager: Arc<DataSyncManager>,
    /// 集群管理器
    cluster_manager: Arc<ClusterManager>,
    /// 当前一致性协议
    protocol: ConsistencyProtocol,
    /// 活跃提案
    active_proposals: Arc<RwLock<HashMap<String, DataProposal>>>,
    /// 已提交提案
    committed_proposals: Arc<RwLock<HashMap<String, DataProposal>>>,
    /// 提案超时时间（秒）
    proposal_timeout: i64,
}

impl ConsistencyManager {
    /// 创建新的一致性管理器
    pub fn new(
        data_sync_manager: Arc<DataSyncManager>,
        cluster_manager: Arc<ClusterManager>,
        protocol: ConsistencyProtocol,
        proposal_timeout: i64,
    ) -> Self {
        Self {
            data_sync_manager,
            cluster_manager,
            protocol,
            active_proposals: Arc::new(RwLock::new(HashMap::new())),
            committed_proposals: Arc::new(RwLock::new(HashMap::new())),
            proposal_timeout,
        }
    }

    /// 提交数据变更提案
    pub async fn propose_change(
        &self,
        sync_message: SyncMessage,
        consistency_level: ConsistencyLevel,
    ) -> Result<String> {
        let healthy_nodes = self.cluster_manager.get_healthy_nodes().await;
        let total_nodes = healthy_nodes.len();
        
        if total_nodes == 0 {
            return Err(NacosError::Cluster("No healthy nodes available".to_string()));
        }

        let required_votes = match consistency_level {
            ConsistencyLevel::Strong => (total_nodes / 2) + 1,
            ConsistencyLevel::Eventual => 1, // Leader approval only
            ConsistencyLevel::ReadCommitted => (total_nodes * 2 / 3) + 1,
            ConsistencyLevel::ReadUncommitted => 1,
        };

        let proposal = DataProposal::new(
            self.cluster_manager.get_current_node().await.map(|n| n.node_id).unwrap_or_else(|| "unknown".to_string()),
            sync_message,
            required_votes as u32,
            self.proposal_timeout,
        );

        let proposal_id = proposal.proposal_id.clone();

        // 存储提案
        {
            let mut proposals = self.active_proposals.write().await;
            proposals.insert(proposal_id.clone(), proposal);
        }

        // 根据协议类型处理
        match self.protocol {
            ConsistencyProtocol::Strong => {
                self.handle_strong_consistency(&proposal_id, consistency_level).await?;
            }
            ConsistencyProtocol::Eventual => {
                self.handle_eventual_consistency(&proposal_id, consistency_level).await?;
            }
            ConsistencyProtocol::Hybrid => {
                self.handle_hybrid_consistency(
                    &proposal_id, 
                    consistency_level
                ).await?;
            }
        }

        Ok(proposal_id)
    }

    /// 处理强一致性
    async fn handle_strong_consistency(
        &self,
        proposal_id: &str,
        _consistency_level: ConsistencyLevel,
    ) -> Result<()> {
        // 使用Raft-like协议
        let proposal = {
            let proposals = self.active_proposals.read().await;
            proposals.get(proposal_id).cloned()
                .ok_or_else(|| NacosError::NotFound(proposal_id.to_string()))?
        };

        // 向所有节点发送投票请求
        let vote_request = VoteRequest {
            vote_id: Uuid::new_v4().to_string(),
            proposal_id: proposal_id.to_string(),
            source_node_id: proposal.proposer_id.clone(),
            version: proposal.sync_message.version,
            timestamp: proposal.sync_message.timestamp,
            data_hash: self.calculate_data_hash(&proposal.sync_message.data_value
                .as_ref()
                .unwrap_or(&Vec::new())
            ),
        };

        // 收集投票
        let mut accepted_votes = 0;
        let healthy_nodes = self.cluster_manager.get_healthy_nodes().await;
        
        for node in healthy_nodes {
            if let Ok(response) = self.send_vote_request(&node, &vote_request).await {
                if matches!(response.status, VoteStatus::Accepted) {
                    accepted_votes += 1;
                }
            }
        }

        // 检查是否获得多数投票
        if accepted_votes >= proposal.required_votes {
            self.commit_proposal(proposal_id).await?;
        } else {
            self.reject_proposal(proposal_id, "Insufficient votes".to_string()).await?;
        }

        Ok(())
    }

    /// 处理最终一致性
    async fn handle_eventual_consistency(
        &self,
        proposal_id: &str,
        _consistency_level: ConsistencyLevel,
    ) -> Result<()> {
        // 领导者直接提交，然后异步传播
        let current_leader = self.cluster_manager.get_leader().await;
        let current_node = self.cluster_manager.get_current_node().await;

        if let (Some(leader), Some(current)) = (current_leader, current_node) {
            if leader.node_id == current.node_id {
                // 当前节点是领导者，直接提交
                self.commit_proposal(proposal_id).await?;
                
                // 异步传播到其他节点
                self.propagate_to_followers(proposal_id).await?;
            } else {
                // 将提案转发给领导者
                self.forward_to_leader(proposal_id, &leader.node_id).await?;
            }
        } else {
            // 无领导者，直接提交到本地
            warn!("No leader available, committing locally");
            self.commit_proposal(proposal_id).await?;
        }

        Ok(())
    }

    /// 处理混合一致性
    async fn handle_hybrid_consistency(
        &self,
        proposal_id: &str,
        consistency_level: ConsistencyLevel,
    ) -> Result<()> {
        match consistency_level {
            ConsistencyLevel::Strong => {
                self.handle_strong_consistency(proposal_id, consistency_level).await?;
            }
            ConsistencyLevel::Eventual => {
                self.handle_eventual_consistency(proposal_id, consistency_level).await?;
            }
            _ => {
                // 默认使用最终一致性
                self.handle_eventual_consistency(proposal_id, consistency_level).await?;
            }
        }

        Ok(())
    }

    /// 发送投票请求
    async fn send_vote_request(
        &self,
        target_node: &ClusterNodeInfo,
        request: &VoteRequest,
    ) -> Result<VoteResponse> {
        // 模拟网络通信 - 实际实现中会通过网络层发送
        let response = VoteResponse {
            vote_id: request.vote_id.clone(),
            responder_id: target_node.node_id.clone(),
            status: VoteStatus::Accepted, // 模拟接受投票
            response_time: Utc::now(),
        };

        Ok(response)
    }

    /// 提交提案
    async fn commit_proposal(&self,
        proposal_id: &str,
    ) -> Result<()> {
        let mut proposals = self.active_proposals.write().await;
        
        if let Some(mut proposal) = proposals.remove(proposal_id) {
            proposal.status = ProposalStatus::Committed;

            // 应用数据变更
            self.data_sync_manager.process_sync_message(
                proposal.sync_message.clone()
            ).await?;

            // 存储已提交的提案
            let mut committed = self.committed_proposals.write().await;
            committed.insert(proposal_id.to_string(), proposal);

            info!("Proposal {} committed", proposal_id);
        }

        Ok(())
    }

    /// 拒绝提案
    async fn reject_proposal(
        &self,
        proposal_id: &str,
        reason: String,
    ) -> Result<()> {
        let mut proposals = self.active_proposals.write().await;
        
        if let Some(mut proposal) = proposals.remove(proposal_id) {
            proposal.status = ProposalStatus::Rejected(reason.clone());
            
            let mut committed = self.committed_proposals.write().await;
            committed.insert(proposal_id.to_string(), proposal);

            warn!("Proposal {} rejected: {}", proposal_id, reason);
        }

        Ok(())
    }

    /// 传播到跟随节点
    async fn propagate_to_followers(
        &self,
        proposal_id: &str,
    ) -> Result<()> {
        let proposal = {
            let committed = self.committed_proposals.read().await;
            committed.get(proposal_id).cloned()
                .ok_or_else(|| NacosError::NotFound(proposal_id.to_string()))?
        };

        // 异步传播到所有健康节点
        self.data_sync_manager.sync_to_all_nodes(
            proposal.sync_message
        ).await?;

        Ok(())
    }

    /// 转发给领导者
    async fn forward_to_leader(
        &self,
        proposal_id: &str,
        leader_id: &str,
    ) -> Result<()> {
        // 实际实现中会通过gRPC转发给领导者
        info!("Forwarding proposal {} to leader {}", proposal_id, leader_id);
        Ok(())
    }

    /// 计算数据哈希
    fn calculate_data_hash(&self, data: &[u8]) -> u64 {
        use std::hash::{Hash, Hasher};
        use std::collections::hash_map::DefaultHasher;
        
        let mut hasher = DefaultHasher::new();
        data.hash(&mut hasher);
        hasher.finish()
    }

    /// 清理过期提案
    pub async fn cleanup_expired_proposals(&self) -> Result<()> {
        let mut proposals = self.active_proposals.write().await;
        let mut to_remove = Vec::new();

        for (proposal_id, proposal) in proposals.iter() {
            if proposal.is_expired() {
                to_remove.push(proposal_id.clone());
            }
        }

        for proposal_id in to_remove {
            if let Some(proposal) = proposals.remove(&proposal_id) {
                let mut committed = self.committed_proposals.write().await;
                committed.insert(proposal_id, proposal);
            }
        }

        Ok(())
    }

    /// 获取提案状态
    pub async fn get_proposal_status(
        &self,
        proposal_id: &str,
    ) -> Option<ProposalStatus> {
        let active = self.active_proposals.read().await;
        if let Some(proposal) = active.get(proposal_id) {
            return Some(proposal.status.clone());
        }

        let committed = self.committed_proposals.read().await;
        if let Some(proposal) = committed.get(proposal_id) {
            return Some(proposal.status.clone());
        }

        None
    }

    /// 获取一致性统计信息
    pub async fn get_consistency_stats(&self,
) -> ConsistencyStats {
        let active = self.active_proposals.read().await;
        let committed = self.committed_proposals.read().await;

        let mut pending_count = 0;
        let mut accepted_count = 0;
        let mut rejected_count = 0;
        let mut committed_count = 0;

        for proposal in active.values() {
            match proposal.status {
                ProposalStatus::Pending => pending_count += 1,
                ProposalStatus::Accepted => accepted_count += 1,
                ProposalStatus::Rejected(_) => rejected_count += 1,
                _ => {},
            }
        }

        for proposal in committed.values() {
            match proposal.status {
                ProposalStatus::Committed => committed_count += 1,
                ProposalStatus::Rejected(_) => rejected_count += 1,
                _ => {},
            }
        }

        ConsistencyStats {
            active_proposals: active.len(),
            committed_proposals: committed.len(),
            pending_proposals: pending_count,
            accepted_proposals: accepted_count,
            rejected_proposals: rejected_count,
            committed_count: committed_count,
        }
    }
}

/// 一致性统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConsistencyStats {
    /// 活跃提案数量
    pub active_proposals: usize,
    /// 已提交提案数量
    pub committed_proposals: usize,
    /// 待处理提案数量
    pub pending_proposals: usize,
    /// 已接受提案数量
    pub accepted_proposals: usize,
    /// 已拒绝提案数量
    pub rejected_proposals: usize,
    /// 已提交计数
    pub committed_count: usize,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::manager::{ClusterConfig, ClusterManager};
    use crate::sync::DataSyncManager;

    #[tokio::test]
    async fn test_proposal_creation() {
        let config = ClusterConfig::default();
        let cluster_manager = Arc::new(ClusterManager::new(config));
        let data_sync_manager = Arc::new(DataSyncManager::new(
            cluster_manager.clone(),
            3,
            100
        ));

        let consistency_manager = ConsistencyManager::new(
            data_sync_manager,
            cluster_manager,
            ConsistencyProtocol::Strong,
            30,
        );

        let sync_message = SyncMessage::new(
            "node1".to_string(),
            "public".to_string(),
            "DEFAULT_GROUP".to_string(),
            "test-key".to_string(),
            crate::sync::SyncType::Incremental,
            crate::sync::DataOperation::Update,
        ).with_data(b"test data".to_vec());

        let proposal = DataProposal::new(
            "node1".to_string(),
            sync_message,
            2,
            30,
        );

        assert_eq!(proposal.required_votes, 2);
        assert!(matches!(proposal.status, ProposalStatus::Pending));
    }

    #[tokio::test]
    async fn test_consistency_manager() {
        let config = ClusterConfig::default();
        let cluster_manager = Arc::new(ClusterManager::new(config));
        let data_sync_manager = Arc::new(DataSyncManager::new(
            cluster_manager.clone(),
            3,
            100
        ));

        let consistency_manager = ConsistencyManager::new(
            data_sync_manager,
            cluster_manager,
            ConsistencyProtocol::Eventual,
            30,
        );

        let sync_message = SyncMessage::new(
            "node1".to_string(),
            "public".to_string(),
            "DEFAULT_GROUP".to_string(),
            "test-key".to_string(),
            crate::sync::SyncType::Incremental,
            crate::sync::DataOperation::Update,
        ).with_data(b"test data".to_vec());

        let proposal_id = consistency_manager.propose_change(
            sync_message,
            ConsistencyLevel::Eventual,
        ).await.unwrap();

        assert!(!proposal_id.is_empty());
        
        // 清理过期提案
        consistency_manager.cleanup_expired_proposals().await.unwrap();
    }
}